package cn.com.lyb.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * DataStream 批处理 单词统计
 */
public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // env.setParallelism(1); // 设置并行度
        // 读取文件
        DataStreamSource<String> lineDS = env.readTextFile("D:\\tool\\idea_code\\lyb_alone\\flink1.17.0\\input\\words.txt");
        // 切分
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    // 转为二元组(work 1)
                    Tuple2<String, Integer> stringIntegerTuple2 = Tuple2.of(word, 1);
                    // 通过采集器向下游发送数据
                    out.collect(stringIntegerTuple2);
                }
            }
        });
        // 分组
        KeyedStream<Tuple2<String, Integer>, String> workAndOneKS = wordAndOne.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        });
        // 聚合 按照(word，cnt)类型
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = workAndOneKS.sum(1);
        // 输出
        sum.print();
        // 执行
        env.execute();


        /*
        输入结果 前面的数字代表并行度
        7> (flink,1)
        3> (hello,1)
        2> (java,1)
        3> (hello,2)
        3> (hello,3)
        5> (world,1)
         */
    }
}
